package com.bravo.advanced.fluent.stream2;

import java.util.ArrayList;
import java.util.List;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;

/**
 * 完整版：包含filter、map、peek、forEach、toList
 *
 * @param <T> 元素类型
 */
public class SimpleStream<T> {
    private final List<T> source;
    private final List<Stage> stages = new ArrayList<>();

    private SimpleStream(List<T> source) {
        this.source = source;
    }

    public static <T> SimpleStream<T> of(List<T> source) {
        return new SimpleStream<>(source);
    }

    public SimpleStream<T> filter(Predicate<? super T> predicate) {
        stages.add(new FilterStage<>(predicate));
        return this;
    }

    @SuppressWarnings("unchecked")
    public <R> SimpleStream<R> map(Function<? super T, ? extends R> mapper) {
        stages.add(new MapStage<>(mapper));
        return (SimpleStream<R>) this;
    }

    public SimpleStream<T> peek(Consumer<? super T> action) {
        stages.add(new PeekStage<>(action));
        return this;
    }

    public void forEach(Consumer<? super T> action) {
        Sink sink = wrapStages(new ForEachSink<>(action));
        for (T element : source) {
            sink.accept(element);
        }
    }

    public List<T> toList() {
        List<T> resultList = new ArrayList<>();
        Sink sink = wrapStages(new ToListSink<>(resultList));
        for (T element : source) {
            sink.accept(element);
        }
        return resultList;
    }

    private Sink wrapStages(Sink sink) {
        for (int i = stages.size() - 1; i >= 0; i--) {
            sink = stages.get(i).wrapSink(sink);
        }
        return sink;
    }

    // Stage：衔接上下游
    private interface Stage {
        Sink wrapSink(Sink downstream);
    }

    // Sink：封装中间操作
    private interface Sink {
        void accept(Object item);
    }

    // 抽象Sink：当前Sink内部包含一个下游Sink，形成链式结构
    private static abstract class ChainedSink implements Sink {
        protected final Sink downstream;

        public ChainedSink(Sink downstream) {
            this.downstream = downstream;
        }
    }

    private static class FilterStage<T> implements Stage {
        private final Predicate<? super T> predicate;

        FilterStage(Predicate<? super T> predicate) {
            this.predicate = predicate;
        }

        @Override
        @SuppressWarnings("unchecked")
        public Sink wrapSink(Sink downstream) {
            // 返回新节点（pre），通过闭包把传入的downstream锁在内部，形成链式结构（sinkNew->sinkOld）
            return new ChainedSink(downstream) {
                @Override
                public void accept(Object item) {
                    // 先执行filter
                    if (predicate.test((T) item)) {
                        // 符合条件，继续downstream操作（比如map）
                        downstream.accept(item);
                    }
                }
            };
        }
    }

    private static class MapStage<T, R> implements Stage {
        private final Function<? super T, ? extends R> mapper;

        MapStage(Function<? super T, ? extends R> mapper) {
            this.mapper = mapper;
        }

        @Override
        @SuppressWarnings("unchecked")
        public Sink wrapSink(Sink downstream) {
            return new ChainedSink(downstream) {
                @Override
                public void accept(Object item) {
                    R result = mapper.apply((T) item);
                    downstream.accept(result);
                }
            };
        }
    }

    private static class PeekStage<T> implements Stage {
        private final Consumer<? super T> action;

        PeekStage(Consumer<? super T> action) {
            this.action = action;
        }

        @Override
        @SuppressWarnings("unchecked")
        public Sink wrapSink(Sink downstream) {
            return new ChainedSink(downstream) {
                @Override
                public void accept(Object item) {
                    action.accept((T) item);
                    downstream.accept(item);
                }
            };
        }
    }

    private static class ForEachSink<T> implements Sink {
        private final Consumer<? super T> action;

        ForEachSink(Consumer<? super T> action) {
            this.action = action;
        }

        @Override
        @SuppressWarnings("unchecked")
        public void accept(Object item) {
            action.accept((T) item);
        }
    }

    private static class ToListSink<T> implements Sink {
        private final List<T> list;

        ToListSink(List<T> list) {
            this.list = list;
        }

        @Override
        @SuppressWarnings("unchecked")
        public void accept(Object item) {
            list.add((T) item);
        }
    }
}
